package models

import (
	"ginclassmates/utils"
	"log"
	"os"

	"github.com/Shopify/sarama"
)

// 创建一个缓冲通道，用于存储要发送的消息
var Messages = make(chan *sarama.ProducerMessage, 100)

// 初始化kafka
func KafkaProduceinit() error {
	config := sarama.NewConfig()
	config.Version = sarama.V3_3_1_0
	config.Producer.Return.Errors = true    // 当消息发送失败时，将错误返回给生产者
	config.Producer.Return.Successes = true // 当消息发送成功时，将成功消息返回给生产者
	config.Producer.Partitioner = sarama.NewHashPartitioner
	sarama.Logger = log.New(os.Stdout, "[Sarama] ", log.LstdFlags)

	// 创建一个新的异步生产者
	p, err := sarama.NewAsyncProducer([]string{utils.Kafkaaddr}, config)
	if err != nil {
		return err
	}
	if p == nil {
		if err != nil {
			return err
		}
	}
	// 定义一个信道，用于接收消息发送结果
	successes := p.Successes()
	// 等待所有消息发送结果
	go func(successes <-chan *sarama.ProducerMessage) {
		for {
			select {
			case success := <-successes:
				log.Printf("Message sent to 分区是 %d at 偏移量 %d\n", success.Partition, success.Offset)
			case err := <-p.Errors():
				log.Println("Error sending message:", err)
				// case <-time.After(time.Second * 30):
				// 	log.Println("Timeout sending message")
			}
		}
	}(successes)
	//开启通道当有消息进入时就发送到kafka中
	go func() {
		for msg := range Messages {
			p.Input() <- msg
		}
	}()
	// defer func() {
	// 	if err := p.Close(); err != nil {
	// 		fmt.Println("Error closing producer:", err)
	// 	}
	// }()
	return nil
}
